package homework

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.graphx.{Edge, Graph, VertexRDD}

/**
  * Created by danke on 2020/4/14.
  */
object Aggregation1 {
  def main(args: Array[String]): Unit = {
    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    // 构建图
    val myVertices = sc.parallelize(Array((1L, 20), (2L, 24), (3L, 18), (4L, 19), (5L, 21),(6L,22)))
    val myEdges = sc.makeRDD(Array(
      Edge(2L, 1L, 7),
      Edge(2L, 4L, 2),
      Edge(3L, 2L, 4),
      Edge(3L, 6L, 3),
      Edge(4L, 1L, 1),
      Edge(5L, 2L, 2),
      Edge(5L, 3L, 8),
      Edge(5L, 6L, 3)
    ))

    val myGraph = Graph(myVertices,myEdges)

    var followers: VertexRDD[(Int,Double)] = myGraph.aggregateMessages[(Int,Double)](
      triplet => {
        if(triplet.srcAttr < triplet.dstAttr){
          triplet.sendToDst((1,triplet.srcAttr))
        }
      },
      (a,b) =>(a._1 +b._1,a._2+b._2)
    )

    println("-------------------------------------------------------------")
    followers.collect.foreach(println(_))
    println("-------------------------------------------------------------")

    val avgAgeOfFollowers:VertexRDD[Double] = followers.mapValues((id,value) =>
    value match {
      case(count,totalAge) => totalAge / count
    })
    avgAgeOfFollowers.collect.foreach(println(_))

  }
}
